package com.times.server.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

/**
 * @version V1.0.0
 * @Title: rabbitmq自定义注入配置bean组件
 * @Package com.times.server.config
 * @Description: ${TODO}
 * @author: TimesPanda
 * @date: 2020-03-31 01:00
 * @Copyright: 2020-03 Inc. All rights reserved.
 * 注意：本内容仅限于公司内部传阅,禁止外泄以及用于其他的商业目
 */
@Configuration
public class RabbitmqConfig {
    //定义日志
    private static final Logger log = LoggerFactory.getLogger(RabbitmqConfig.class);

    //自动装配rabbitmq的连接工厂配置
    @Autowired
    private CachingConnectionFactory connectionFactory;
    //自动装配消息监听器所在的容器工厂配置类实例
    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;
    @Autowired
    private Environment env;

    /**
     * 单一消费者配置实例
     * @return
     */
    @Bean(name="simpleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer(){
        //定义消息监听器所在的容器工厂
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        //设置容器工厂所用的实例
        factory.setConnectionFactory(connectionFactory);
        //设置消息传输的格式,这里采用json格式传输
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //设置并发消费者实例的初始数量,这里为1
        factory.setConcurrentConsumers(1);
        //设置并发消费者实例的最大数量,这里为1
        factory.setMaxConcurrentConsumers(1);
        //设置并发消费者实例中每个实例拉取的消息数量,这里为1
        factory.setPrefetchCount(1);
        return factory;
    }
    /**
     * 多消费者配置实例,针对并发业务的操作配置
     * @return
     */
    @Bean(name="multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer(){
        //定义消息监听器所在的容器工厂
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        //设置容器工厂所用的实例
        factoryConfigurer.configure(factory,connectionFactory);//单一场景：factory.setConnectionFactory(connectionFactory);
        //设置消息传输的格式,这里采用json格式传输
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //(增加)设置消息的确认消费模式,在这里为none,表示不需要确认消费
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        //设置并发消费者实例的初始数量,这里为10
        factory.setConcurrentConsumers(10);
        //设置并发消费者实例的最大数量,这里为15
        factory.setMaxConcurrentConsumers(15);
        //设置并发消费者实例中每个实例拉取的消息数量,这里为10
        factory.setPrefetchCount(10);
        return factory;
    }

    /**
     * 自定义配置rabbitmq发送消息的操作组件rabbitTemplate
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate(){
        //设置"发送消息后进行确认"
        connectionFactory.setPublisherConfirms(true);
        //设置"发送消息后返回确认信息"
        connectionFactory.setPublisherReturns(true);
        //构造发送消息组件实例
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        //发送消息后,如果发送成功,则输出“消息发送成功”的反馈信息
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("消息发送成功：correlationData({}),ack({}),cause({})",ack,cause);
            }
        });
        //发送消息后,如果发送失败,则输出“消息发送失败”的反馈信息
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("消息丢失：exchange({}),route({}),replyCode({}),replyText({}),message({})",exchange,routingKey,replyCode,replyText,message);
            }
        });
        return rabbitTemplate;
    }
    /**
     * 创建简单消息模型：队列、交换机和路由
     * @return
     */
    //创建队列
    @Bean(name="basicQueue")
    public Queue basicQueue() {
        return new Queue(env.getProperty("mq.basic.info.queue.name"), true);
    }
    //创建交换机：在这里以DirectExchange为例
    @Bean
    public DirectExchange basicExchange() {
        return new DirectExchange(env.getProperty("mq.basic.info.exchange.name"), true, false);
    }
    //创建绑定
    @Bean
    public Binding basicBinding(){
        return BindingBuilder.bind(basicQueue()).to(basicExchange()).with(env.getProperty("mq.basic.info.routing.key.name"));
    }
}
